<?php
/**
 * Created by PhpStorm.
 * User: sxg
 * Date: 2022/8/1
 * Time: 14:24
 */

require_once "vendor/autoload.php";

ini_set("memory_limit", "2048M");

use Socket\Ms\Protocols\Mqtt;
use Socket\Ms\Response;
use Socket\Ms\TcpConnections;

class ms {

    private $_server;

    public function __construct() {
        $this->_server = new Socket\Ms\Server("mqtt://0.0.0.0:9501");
//        $this->_server->on("connect", [$this, "onConnect"]);
//        $this->_server->on("request", [$this, 'onRequest']);
//        $this->_server->on("receive", [$this, "onReceive"]);
        $this->_server->settings([
            'workerNum' => 1,
            'unix_server_socket_file' => '/home/sxg/te/sock/unix_sock_server.sock',
            'unix_client_socket_file' => '/home/sxg/te/sock/unix_sock_client.sock',
            'daemon' => false
        ]);
//        $this->_server->on("masterStart", [$this, "masterStart"]);
//        $this->_server->on("masterShutdown", [$this, "masterShutdown"]);
        $this->_server->on("workerStart", [$this, "workerStart"]);
//        $this->_server->on("workerStop", [$this, "workerStop"]);
//        $this->_server->on("workerReload", [$this, "workerReload"]);
//        $this->_server->on("task", [$this, "task"]);
//        $this->_server->on("open", [$this, "onOpen"]);
        $this->_server->on("open", [$this, "open"]);
        $this->_server->on("subscribe", [$this, "subscribe"]);
        $this->_server->on("unsubscribe", [$this, "unsubscribe"]);
        $this->_server->on("publish", [$this, "publish"]);
//        $this->_server->on("message", [$this, "onMessage"]);
        $this->_server->on("close", [$this, "onClose"]);
        $this->_server->start();
    }
//
//    public function onConnect(Socket\Ms\Server $server, Socket\Ms\TcpConnections $TcpConnections) {
////        $server->echoLog("有客户端连接了");
//    }

//    public function onRequest(\Socket\Ms\Server $server, \Socket\Ms\Request $request, Response $response): bool
//    {
//        if (preg_match("/.jpg|.html|.png|.gif|.js|jpeg/", $_REQUEST['uri'])) {
//            $response->sendFile('www/'.$_REQUEST['uri']);
//            return true;
//        }
////        $response->setHeader("Content-Type", "application/json");
////        $response->write(json_encode(['name' => 'sxg', 'age' => '25']));
//        global $routes;
//        global $dispatch;
//        $dispatch->callAction($routes, $request, $response);
//        return true;
//    }

//    public function onReceive(Socket\Ms\Server $server, $msg, Socket\Ms\TcpConnections $connection) {
//        $server->echoLog("有客户发送数据:%s", $msg);
//        $server->echoLog(time());
//        if (DIRECTORY_SEPARATOR == '/') {
//            $server->task(function ($result) use ($server) {
//                sleep($result);
//                $server->echoLog( "异步任务执行了");
//                $server->echoLog(time());
//            });
//        }
//        $connection->send("i am a server ".time());
//    }
//
//    public function onOpen(Socket\Ms\Server $server, Socket\Ms\TcpConnections $connection) {
//        $server->echoLog("握手成功");
//    }

    public function open(Socket\Ms\Server $server, Socket\Ms\TcpConnections $connection) {
        echo "有客户端连接上来了\n";
        /**@var $proObj Mqtt*/
        $proObj = $connection->protocol;
//        print_r($proObj);
        return true;
    }

    public function subscribe(Socket\Ms\Server $server, Socket\Ms\TcpConnections $connection) {
        $proObj = $connection->protocol;
        print_r($proObj->subscribeTopics);
    }

    public function unsubscribe(Socket\Ms\Server $server, Socket\Ms\TcpConnections $connection) {
        $proObj = $connection->protocol;
        print_r($proObj->subscribeTopics);
    }

    public function publish(Socket\Ms\Server $server, Socket\Ms\TcpConnections $connection) {
        $server->echoLog("接收发布报文了,发布主题:%s,发布消息:%s", $connection->protocol->publishPackTopicName, $connection->protocol->publishPacketMessage);
        //需要给所有订阅的客户端转发
        /**@var $proObj Mqtt*/
        $proObj = $connection->protocol;
//        foreach ($server::$_connections as $conn) {
//            if ($conn->protocol instanceof Mqtt) {
//                if (isset($conn->protocol->subscribeTopics[$connection->protocol->publishPackTopicName])) {
//                    $publishData = $proObj->responsePUBLISHPACK($conn->protocol->subscribeTopics[$connection->protocol->publishPackTopicName]);
//                    $conn->sendDataForMqttPublish($publishData);
//                }
//            }
//        }

        //转发数据给websocket客户端
        foreach ($server::$_connections as $client) {
            /**@var $client TcpConnections*/
            if ($client->protocol instanceof \Socket\Ms\Protocols\WebSocket) {
                if ($client->subscribeTopic == $proObj->publishPackTopicName) {
                    $client->send(json_encode(['code' => 300, 'esp32' => $proObj->publishPacketMessage], JSON_UNESCAPED_UNICODE));
                }
            }
        }
    }

//    public function onMessage(Socket\Ms\Server $server, Socket\Ms\TcpConnections $connection, $frame) {
//        $server->echoLog("有客户端发送数据:%s", $frame);
//        $connection->send("是呀是呀");
//    }

    public function onClose(Socket\Ms\Server $server, Socket\Ms\TcpConnections $connection) {
//        $server->echoLog("connection close");
        $server->removeClient($connection->_socketFd);
    }

    public function masterStart(Socket\Ms\Server $server) {
        $server->echoLog("master server <pid:%d> start working", posix_getpid());
    }

    public function masterShutdown(Socket\Ms\Server $server) {
        $server->echoLog("master server <pid:%d> shutdown", posix_getpid());
    }

    public function workerStart(Socket\Ms\Server $server) {
        global $routes;
        global $dispatch;
        $routes = require_once "app/routes/api.php";
        $dispatch = new \App\Controller\DispatchController();

        //mqtt服务中添加http服务
        $httpServer = new Socket\Ms\Server("http://0.0.0.0:9502");
        $httpServer->on("request", function (\Socket\Ms\Server $server, \Socket\Ms\Request $request, Response $response) {
            if (preg_match("/.jpg|.html|.png|.gif|.js|jpeg/", $_REQUEST['uri'])) {
                $response->sendFile('www/'.$_REQUEST['uri']);
                return true;
            }
            global $routes;
            global $dispatch;
            //收到客户端的请求后给websocket客户端发消息
            foreach ($server::$_connections as $connection) {
                if ($connection->protocol instanceof \Socket\Ms\Protocols\WebSocket) {
                    $connection->send(json_encode($request->_get, JSON_UNESCAPED_UNICODE));
                }
            }
            $dispatch->callAction($routes, $request, $response);
            return true;
        });
        $httpServer->listen();
        $httpServer->acceptClient();

        //mqtt服务中添加websocket服务
        $wsServer = new \Socket\Ms\Server("ws://0.0.0.0:9503");
        $wsServer->on("open", function (Socket\Ms\Server $server, Socket\Ms\TcpConnections $TcpConnections) {
            echo "有客户端连接上来了\r\n";
        });
        $wsServer->on("message", function (Socket\Ms\Server $server, Socket\Ms\TcpConnections $connection, $msg) {
            //收到客户端的请求后给websocket客户端发消息
            echo "接收到客户端的消息:".$msg."\r\n";
            global $dispatch;
            global $routes;
            $dispatch->callWsAction($routes, $msg, $connection);
        });
        $wsServer->listen();
        $wsServer->acceptClient();

        echo "\e[33;40mHttp server listen on 0.0.0.0:9502\e[0m\r\n";
        echo "\e[33;40mWs  server listen on 0.0.0.0:9503\e[0m\r\n";
    }

    public function workerStop(Socket\Ms\Server $server) {
        $server->echoLog("worker <pid:%d> stop", posix_getpid());
    }

    public function workerReload(Socket\Ms\Server $server) {
        $server->echoLog("worker <pid:%d> reload", posix_getpid());
    }

    public function task(Socket\Ms\Server $server, $msg) {
        $server->echoLog("task process <pid:%d> recv msg:%s", posix_getpid(), $msg);
    }
}

$ms = new ms();

